package com.example.hadoop;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.compress.utils.IOUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.springframework.stereotype.Component;

import com.example.entity.FileInfo;

/**
 * 
 * HDFS(分布式文件存储系统操作类).
 * 
 * 
 * 
 * @author hjn
 * 
 * @version 1.0 2013-11-22
 */
@Component
public class HDFS {

	/**
	 * 
	 * 文件系统类
	 */

	private FileSystem fileSystem;

	/**
	 * 
	 * 无参数构造方法
	 */

	public HDFS() {

		init();

	}

	/**
	 * 
	 * 初始化
	 */

	private void init() {

		fileSystem = HDFSServer.getFileSystem();

	}
	
	/**
	 * 
	 * 创建HDFS目录
	 * 
	 * 
	 * 
	 * @param dir
	 */

	public boolean createDir(String dir) throws IOException {

		Path path = new Path(dir);

		if (fileSystem.exists(path)) {

			System.out.println("此目录已经存在不需要再创建");

			return true;

		}

		return fileSystem.mkdirs(path);

	}
	
	/**
	 * 
	 * HDFS创建文件
	 * 
	 * 
	 * 
	 * @param in输入流
	 * 
	 * @param dst分布式存储路径
	 * 
	 * @throws IOException
	 */

	public void create(InputStream in, String dst) throws IOException {

		Path dstPath = new Path(dst);

		FSDataOutputStream out = fileSystem.create(dstPath);

		IOUtils.copy(in, out);

	}


	/**
	 * 
	 * 在HDFS创建文件
	 * 
	 * 
	 * 
	 * @param file
	 * 
	 * @param dst分布式存储路径
	 * 
	 * @throws IOException
	 */

	public void create(File file, String dst) throws IOException {

		InputStream in = new FileInputStream(file);

		this.create(in, dst);

	}

	/**
	 * 
	 * 在HDFS创建文件
	 * 
	 * 
	 * 
	 * @param src本地文件路径
	 * 
	 * @param dst分布式存储路径
	 * 
	 * @throws IOException
	 */

	public void create(String src, String dst) throws IOException {

		File file = new File(src);

		this.create(file, dst);

	}
	
	
	public void createFile(String content,String remoteFilePath) throws IOException {  
  
        FSDataOutputStream os = fileSystem.create(new Path(remoteFilePath));  
  
        os.write(content.getBytes("UTF-8"));  
  
        os.close();  
  
    }  
	
	/**
	  * 创建文件内容
	  */
	 public void createToFile(String localFilePath, String remoteFilePath) throws IOException {
	     Path remotePath = new Path(remoteFilePath);
	     /* 创建一个文件读入流 */
	     FileInputStream in = new FileInputStream(localFilePath);
	     /* 创建一个文件输出流，输出的内容将追加到文件末尾 */
	     FSDataOutputStream out = fileSystem.create(remotePath);
	     /* 读写文件内容 */
	     byte[] data = new byte[1024];
	     int read = -1;
	     while ( (read = in.read(data)) > 0 ) {
	         out.write(data, 0, read);
	     }
	     out.close();
	     in.close();
	 }
	 
	/**
	  * 追加文件内容
	  */
	 public void appendToFile(String localFilePath, String remoteFilePath) throws IOException {
	     Path remotePath = new Path(remoteFilePath);
	     /* 创建一个文件读入流 */
	     FileInputStream in = new FileInputStream(localFilePath);
	     /* 创建一个文件输出流，输出的内容将追加到文件末尾 */
	     FSDataOutputStream out = fileSystem.append(remotePath);
	     /* 读写文件内容 */
	     byte[] data = new byte[1024];
	     int read = -1;
	     while ( (read = in.read(data)) > 0 ) {
	         out.write(data, 0, read);
	     }
	     out.close();
	     in.close();
	 }
	
	

	/**
	 * 
	 * 获取指定目录列表路径
	 * 
	 * 
	 * 
	 * @param dirPath
	 */

	public List<String> dir(String dirPath) throws IOException {

		List<String> fileList = null;

		Path path = new Path(dirPath);

		if (fileSystem.exists(path)) {

			fileList = new ArrayList<String>();

			FileStatus[] list = this.getFileStatus(path);

			for (FileStatus fileStatus : list) {

				fileList.add(fileStatus.getPath().toString());

			}

		} else {

			System.out.println("目录不存在");

		}

		return fileList;

	}
	
	public FileStatus[] dires(String dirPath) throws IOException {

		Path path = new Path(dirPath);

		if (fileSystem.exists(path)) {

			return this.getFileStatus(path);

		} else {

			System.out.println("目录不存在");

		}

		return null;

	}
	
	public List<FileInfo> dirFile(String dirPath) throws IOException {

		List<FileInfo> fileList = null;

		Path path = new Path(dirPath);

		if (fileSystem.exists(path)) {

			fileList = new ArrayList<FileInfo>();

			FileStatus[] list = this.getFileStatus(path);

			for (FileStatus fileStatus : list) {
				FileInfo file = new FileInfo();
				file.setBlocksize(fileStatus.getBlockSize());
				file.setDirectory(fileStatus.isDirectory());
				file.setLength(fileStatus.getLen());
				file.setModificationTime(fileStatus.getModificationTime());
				file.setName(fileStatus.getPath().getName());
				file.setPath(fileStatus.getPath().getName());
				fileList.add(file);
			}

		} else {

			System.out.println("目录不存在");

		}

		return fileList;

	}
	
	/**
	 * 
	 * 递归获取指定目录列表路径
	 * 
	 * 
	 * 
	 * @param dirPath
	 */

	public List<String> dirs(String dirPath) throws IOException {

		List<String> fileList = null;

		Path path = new Path(dirPath);

		if (fileSystem.exists(path)) {

			fileList = new ArrayList<String>();

			// listFiles列出的是文件信息，而且提供递归遍历
			RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(path, true);
			
			while(files.hasNext()){
				
				LocatedFileStatus file = files.next();
				Path filePath = file.getPath();
				String fileName = filePath.getName();
				System.out.println(fileName);
				
			}
			
			System.out.println("---------------------------------");
			
			//listStatus 可以列出文件和文件夹的信息，但是不提供自带的递归遍历
			FileStatus[] listStatus = fileSystem.listStatus(path);
			for(FileStatus status: listStatus){
				
				String name = status.getPath().getName();
				System.out.println(name + (status.isDirectory()?" is dir":" is file"));
				fileList.add(status.getPath().toString());
			}

		} else {

			System.out.println("目录不存在");

		}

		return fileList;

	}
	
	/**
	 * 
	 * HDFS文件是否存在
	 * 
	 * 
	 * 
	 * @param filePath
	 * 
	 * @return boolean:是否存在
	 * 
	 * @throws IOException
	 */

	public boolean exists(String filePath) throws IOException {

		Path path = new Path(filePath);

		return fileSystem.exists(path);

	}
	
	public boolean isDir(String src) throws IOException {

		Path path = new Path(src);

		return fileSystem.isDirectory(path);

	}
	
	 /**
	  * 读取文件内容
	  */
	 public String cat(String remoteFilePath) throws IOException {
		 String message = "";
	     Path remotePath = new Path(remoteFilePath);
	     FSDataInputStream in = fileSystem.open(remotePath);
	     BufferedReader d = new BufferedReader(new InputStreamReader(in));
	     String line = null;
	     while ( (line = d.readLine()) != null ) {
//	         System.out.println(line);
	    	 message = line;
	     }
	    d.close();
	    in.close();
	    return message;
	 }

	/**
	 * 
	 * 获取文件
	 * 
	 * 
	 * 
	 * @param filePath
	 * 
	 * @return
	 * 
	 * @throws IOException
	 */

	public InputStream getFile(String filePath) throws IOException {

		Path path = new Path(filePath);

		return fileSystem.open(path);

	}

	/**
	 * 
	 * 获取HDFS指定目录下文件状态列表
	 * 
	 * 
	 * 
	 * @param dirPath指定目录路径
	 * 
	 * @return fileStatusList
	 * 
	 * @throws FileNotFoundException
	 * 
	 * @throws IOException
	 */

	public FileStatus[] getFileStatus(Path path) throws FileNotFoundException,IOException {

		FileStatus[] fileStatusList = fileSystem.listStatus(path);

		return fileStatusList;

	}
	
	/**
	 * 
	 * 更改HDSF文件名称
	 * 
	 * 
	 * 
	 * @param fileOldName
	 * 
	 * @param fileNewName
	 * 
	 * @return boolean:是否更名字成功
	 * 
	 * @throws IOException
	 */

	public boolean rename(String src, String dst) throws IOException {

		Path srcPath = new Path(src);

		if (fileSystem.exists(srcPath)) {

			Path dstPath = new Path(dst);

			return fileSystem.rename(srcPath, dstPath);

		}

		System.out.println("原文件不存在");

		return false;

	}



	/**
	 * 
	 * 上传本地文件到HDFS（注意是服务器本地硬盘，非客户端硬盘)）
	 * 
	 * 
	 * 
	 * @return
	 * 
	 * 
	 * 
	 * @throws IOException
	 */

	public void uploadLocalFile(String localFileSrc, String HDFSFileDst)
			throws IOException {

		Path src = new Path(localFileSrc);

		Path dst = new Path(HDFSFileDst);

		fileSystem.copyFromLocalFile(src, dst);

	}

	/**
	 * 
	 * 批量上传本地文件到HDFS
	 * 
	 * @param localFileSrcs本地文件列表
	 * 
	 * @param HDFSFileDst
	 * 
	 * @throws IOException
	 */

	public void uploadLocalFile(String[] localFileSrcs, String HDFSFileDst)
			throws IOException {

		Path dstPath = new Path(HDFSFileDst);

		Path[] paths = new Path[localFileSrcs.length];

		for (int i = 0; i < localFileSrcs.length; i++) {

			paths[i] = new Path(localFileSrcs[i]);

		}

		fileSystem.copyFromLocalFile(false, false, paths, dstPath);

	}

	/**
	 * 
	 * 从HDFS下载文件到本地(注意是服务器本地硬盘，非浏览器客户端硬盘)
	 * 
	 * 
	 * 
	 * @param HDFSFilePath
	 * 
	 * @param localFilePath
	 * 
	 * @throws IOException
	 */

	public void downFileToLocal(String HDFSFilePath, String localFilePath)

	throws IOException {

		Path dstPath = new Path(HDFSFilePath);

		FSDataInputStream in = fileSystem.open(dstPath);

		OutputStream out = new FileOutputStream(new File(localFilePath));

		IOUtils.copy(in, out);

	}



	/**
	 * 
	 * 根据路径删除文件或文件夹
	 * 
	 * 
	 * 
	 * @param filePath
	 * 
	 * @return
	 * 
	 * @throws IOException
	 */

	@SuppressWarnings("deprecation")
	public boolean deleteFile(String filePath) throws IOException {

		if (this.exists(filePath)) {

			Path path = new Path(filePath);

			fileSystem.delete(path);

			return true;

		}

		System.out.println("文件不存在");

		return false;

	}

	/**
	 * 
	 * 剪切本地文件到HDFS(注意为服务器本地文件);
	 * 
	 * 
	 * 
	 * @param src本地路径
	 * 
	 * @param dst分布式存储路径
	 * 
	 * @throws IOException
	 */

	public void moveFromLocalFile(String localSrc, String HDFSDst)
			throws IOException {

		Path srcPath = new Path(localSrc);

		Path dstPath = new Path(HDFSDst);

		fileSystem.moveFromLocalFile(srcPath, dstPath);

	}

	/**
	 * 
	 * HDFS文件之间的复制
	 * 
	 * @param src源文件路径
	 * 
	 * @param dst要复制后复制文件的路径
	 * 
	 * @throws IOException
	 */

	public void copyHDFSFile(String src, String dst) throws IOException {

		Path srcPath = new Path(src);

		Path dstPath = new Path(dst);

		InputStream in = fileSystem.open(srcPath);

		OutputStream out = fileSystem.create(dstPath);

		IOUtils.copy(in, out);

	}

	/**
	 * 
	 * HDFS中移动文件
	 * 
	 * @param src源文件路径
	 * 
	 * @param dst要移动后的路径
	 * 
	 * @throws IOException
	 */

	@SuppressWarnings("deprecation")
	public void moveHDFSFile(String src, String dst) throws IOException {

		Path srcPath = new Path(src);

		Path dstPath = new Path(dst);

		InputStream in = fileSystem.open(srcPath);

		OutputStream out = fileSystem.create(dstPath);

		IOUtils.copy(in, out);

		fileSystem.delete(srcPath);

	}

	/**
	 * 
	 * 剪切HDFS文件到本地
	 * 
	 * @param HDFSSrc
	 * 
	 * @param localDst
	 * 
	 * @throws IOException
	 */

	public void moveToLocalFile(String HDFSSrc, String localDst)
			throws IOException {

		Path srcPath = new Path(HDFSSrc);

		Path dstPath = new Path(localDst);

		fileSystem.moveToLocalFile(srcPath, dstPath);

	}

	
	/**
	 * 从hdfs上下载文件
	 * @Name download
	 * @Parameters @param src  hdfs路径
	 * @Parameters @param dst  本地路径
	 * @Parameters @throws IOException
	 * @Return void
	 * @Throws
	 */
	public void download(String src, String dst) throws IOException {
		fileSystem.copyToLocalFile(new Path(src), new Path(dst));
	}

	/**
	 * 
	 * 获取FileSystem对象
	 * 
	 * 
	 * 
	 * @return
	 */

	public FileSystem getFileSystem() {

		return fileSystem;

	}

	/**
	 * 
	 * 关闭HDFS
	 * 
	 * 
	 * 
	 * @throws IOException
	 */

	public void close() throws IOException {

		fileSystem.close();

	}


	public static void main(String[] args) throws IOException {

		
		
		HDFS hdfs = new HDFS();
		
		//获取文件
		System.out.println(hdfs.dirs("/"));
//		System.out.println(hdfs.dir("/demo"));
		
		//是否存在目录或者文件
//		 System.out.println(hdfs.exists("/demo/2/QQͼƬ20170524153300.png"));

		//上传本地目录
//		 hdfs.uploadLocalFile("d:/compressed.tracemonkey-pldi-09.pdf","/pdf/compressed.tracemonkey-pldi-09.pdf");
//		 hdfs.createToFile("d:/compressed.tracemonkey-pldi-09.pdf","/compressed.tracemonkey-pldi-09.pdf");
//		 hdfs.uploadLocalFile("D:/upload", "/test"); 
//		 hdfs.uploadLocalFile(new String[]{"d:/home/www/images","d:/home/www/ixiaofu"}, "/");
		
		//重命名
//		 hdfs.rename("/test", "/demo");

		 //删除文件
//		 hdfs.deleteFile("/ixiaofu");

		 // 剪切本地文件到HDFS
//		 hdfs.moveFromLocalFile("‪‪D:/www/test.tar","/pdf/test.tar");
		
		//创建文件
//		hdfs.createDir("/test");
		
		//下载文件
//		hdfs.downFileToLocal("/test.log", "d:/test.log");
		
//		//复制文件
//		hdfs.copyHDFSFile("/test.log", "/demo/test.log");
//		
//		//移动文件
//		hdfs.moveHDFSFile("/test.log", "/user/test.log");
		
//		System.out.println("***************"+hdfs.cat("/test1.log"));

		hdfs.close();

	}

}